package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/24 11:39
 */
public class Dwd_14_InteractionFavorAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_14_InteractionFavorAdd().init(2014, 2, "Dwd_14_InteractionFavorAdd");
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        
        readOdsDb(tEnv, "Dwd_14_InteractionFavorAdd");
    
        Table resultTable = tEnv.sqlQuery("select " +
                                        "data['id'] id, " +
                                        "data['user_id'] user_id, " +
                                        "data['sku_id'] sku_id, " +
                                        "date_format(data['create_time'],'yyyy-MM-dd') date_id, " +
                                        "data['create_time'] create_time, " +
                                        "ts " +
                                        "from ods_db " +
                                        "where `table`='favor_info' " +
                                        "and `type`='insert' ");
        
        tEnv.executeSql("create table dwd_interaction_favor_add ( " +
                            "id string, " +
                            "user_id string, " +
                            "sku_id string, " +
                            "date_id string, " +
                            "create_time string, " +
                            "ts string " +
                            ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_INTERACTION_FAVOR_ADD));
    
        resultTable.executeInsert("dwd_interaction_favor_add");
    
    
    }
}
